﻿
namespace CSharp_RabbitMQ_Demo
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    using RabbitMQ.Client;
    using System.IO;
    using System.Collections.Concurrent;
    public class RabbitMQConsumer : DefaultBasicConsumer
    {
        public ConcurrentQueue<string> Queue
        {
            get;
            protected set;
        }

        public RabbitMQConsumer(IModel model) : this(model, new ConcurrentQueue<string>())
        {
            this.Queue = new ConcurrentQueue<string>();
        }


        public RabbitMQConsumer(IModel model, ConcurrentQueue<string> queue) : base(model)
        {
            this.Queue = queue;
        }


        public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body)
        {
            var o = Encoding.UTF8.GetString(body);
            this.Queue.Enqueue(o);
        }

        public override void OnCancel()
        {
            base.OnCancel();
        }

        public string TryDequeue()
        {
            string m;
            if (!Queue.TryDequeue(out m))
            {
                if (!Model.IsOpen)
                    throw new EndOfStreamException();
            }
            return m;
        }
    }
}